package com.zxg.util.concurrent.multi;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zxg.zbc.dal.dao.concurrent.ConcurrentDao;
import com.zxg.zbc.dal.dataobject.concurrent.LastProcessed;
import com.zxg.zbc.dal.dataobject.concurrent.Lock;
import com.zxg.zbc.dal.dataobject.concurrent.LastProcessed.DataGetType;

public class BasicProcessJob<T> {
  private final static Logger         log                   = LoggerFactory.getLogger(BasicProcessJob.class);

  /**
   * 线程池大小
   */
  private Integer                     taskPoolSize          = 10;
  /**
   * 队列大小 0为无限大不限制
   */
  private Integer                     taskQueenSize         = 0;
  /**
   * hostid
   */
  public static final String          HOST_ID;
  /**
   * 获取数据时每次获取的个数
   */
  private Integer                     itemLoadCount         = 20;
  /**
   * 每个task包含的要处理的宝贝个数
   */
  private Integer                     everyTaskItemCount    = 4;
  /**
   * 线程池满等待时间
   */
  private Long                        taskPoolFullWaiteTime = 10000L;
  /**
   * 默认等待时间
   */
  private Integer                     defaultWaitTime       = 2000;
  /**
   * 默认最大等待时间
   */
  private Integer                     defaultMaxWaiteTime   = 5 * 60 * 1000;

  private static final Random         random                = new Random();

  /**
   * 业务相关 锁的type
   */
  private final Lock.ConcurrentType   type;

  private DataGetType                 dataGetType;
  /*
   * 执行器 获取数据 处理数据
   */
  private BasicProcessTaskExecutor<T> executor;
  private ThreadPoolExecutor          taskThreadPool;

  private ConcurrentDao               concurrentDao;

  public BasicProcessJob(Lock.ConcurrentType type, DataGetType dataGetType, BasicProcessTaskExecutor<T> executor, ConcurrentDao concurrentDao) {
    super();
    this.type = type;
    this.dataGetType = dataGetType;
    this.executor = executor;
    this.concurrentDao = concurrentDao;
    if (taskQueenSize != null && taskQueenSize > 0)
      taskThreadPool = new ThreadPoolExecutor(taskPoolSize, taskPoolSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(taskQueenSize));
    else
      taskThreadPool = new ThreadPoolExecutor(taskPoolSize, taskPoolSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
  }

  static {
    String hostId = null;
    try {
      hostId = InetAddress.getLocalHost().getHostName();
    } catch (Exception e) {
    }
    HOST_ID = hostId;
  }

  public void exec() {

    boolean sleep = false;

    int sleepMiliSecs = defaultWaitTime;

    do {

      if (sleep) {

        // log.info("sleep: " + sleepMiliSecs + "miliSec");
        try {
          if (sleepMiliSecs >= defaultMaxWaiteTime) {
            sleepMiliSecs = defaultMaxWaiteTime;
          }
          Thread.sleep(sleepMiliSecs);
          sleep = false;
        } catch (Exception e) {
          log.error("Thread.sleep ", e);
        }
      }

      // log.warn("DeleteItemSnapeProcessJob: " +
      // System.currentTimeMillis());

      // 加鎖逻辑
      try {
        if (!concurrentDao.lock(HOST_ID, type, 600)) {
          // 别人持有锁 睡一会
          random.setSeed(System.currentTimeMillis());
          sleepMiliSecs = random.nextInt(4000) + 1000;
          sleep = true;
          continue;
        }
      } catch (Exception e) {
        log.error("get lock exception.", e);
        sleepMiliSecs = defaultWaitTime;
        sleep = true;
        continue;
      }

      BasicProcessDatas<T> datas = null;

      // 获取数据并且释放锁逻辑
      try {
        // 获取数据逻辑
        LastProcessed from = concurrentDao.getLastProcessedByType(type);
        if (dataGetType.equals(DataGetType.ID))
          datas = executor.loadDatas(from.getLastProcessId(), this.itemLoadCount);
        else
          datas = executor.loadDatas(from.getLastProcessTime(), this.itemLoadCount);

        if (datas == null) {
          random.setSeed(System.currentTimeMillis());
          sleepMiliSecs = random.nextInt(4000) + 1000;
          sleep = true;
          continue;
        }

        from.setLastProcessHost(HOST_ID);

        if (dataGetType.equals(DataGetType.ID)) {
          from.setLastProcessId(datas.getUpdateId());
        } else {
          from.setLastProcessTime(datas.getUpdateTime());
        }
        concurrentDao.updateLastProcessed(from);
        log.warn("updateLastProcesss Id:" + from.getLastProcessId());
        if (datas.getDatas().size() == 0) {
          log.warn("datas.getDatas().size() == 0 continue");
          random.setSeed(System.currentTimeMillis());
          sleepMiliSecs = random.nextInt(4000) + 1000;
          sleep = true;
          continue;
        }

      } catch (Exception e) {
        log.error("error:", e);
        sleep = true;
        sleepMiliSecs = defaultWaitTime;
        continue;
      } finally {
        try {
          if (!concurrentDao.unlock(HOST_ID, type)) {

          }
        } catch (Exception e2) {
          // ignore me.
        }
      }

      // 处理分发任务逻辑
      int times = datas.getDatas().size() / everyTaskItemCount;
      if (datas.getDatas().size() % everyTaskItemCount != 0) {
        times = times + 1;
      }
      for (int i = 0; i < times;) {
        int start = i * everyTaskItemCount;
        int to = (i + 1) * everyTaskItemCount > datas.getDatas().size() ? datas.getDatas().size() : (i + 1) * everyTaskItemCount;
        List<T> t = new ArrayList<T>(datas.getDatas().subList(start, to));
        try {
          taskThreadPool.execute(new BasicProcessTask<T>(t, executor));
          i++;
        } catch (Exception e) {
          log.error("taskThreadPool.execute error:", e);
          try {
            Thread.sleep(taskPoolFullWaiteTime);
          } catch (InterruptedException e1) {
            continue;
          }
          continue;
        }
      }
      // 设置下次循环等待模式
      sleep = true;
      sleepMiliSecs = defaultWaitTime;

    } while (true);
  }

  public Integer getTaskPoolSize() {
    return taskPoolSize;
  }

  public void setTaskPoolSize(Integer taskPoolSize) {
    this.taskPoolSize = taskPoolSize;
  }

  public Integer getTaskQueenSize() {
    return taskQueenSize;
  }

  public void setTaskQueenSize(Integer taskQueenSize) {
    this.taskQueenSize = taskQueenSize;
  }

  public Integer getItemLoadCount() {
    return itemLoadCount;
  }

  public void setItemLoadCount(Integer itemLoadCount) {
    this.itemLoadCount = itemLoadCount;
  }

  public Integer getEveryTaskItemCount() {
    return everyTaskItemCount;
  }

  public void setEveryTaskItemCount(Integer everyTaskItemCount) {
    this.everyTaskItemCount = everyTaskItemCount;
  }

  public Long getTaskPoolFullWaiteTime() {
    return taskPoolFullWaiteTime;
  }

  public void setTaskPoolFullWaiteTime(Long taskPoolFullWaiteTime) {
    this.taskPoolFullWaiteTime = taskPoolFullWaiteTime;
  }

  public Integer getDefaultWaitTime() {
    return defaultWaitTime;
  }

  public void setDefaultWaitTime(Integer defaultWaitTime) {
    this.defaultWaitTime = defaultWaitTime;
  }

  public Integer getDefaultMaxWaiteTime() {
    return defaultMaxWaiteTime;
  }

  public void setDefaultMaxWaiteTime(Integer defaultMaxWaiteTime) {
    this.defaultMaxWaiteTime = defaultMaxWaiteTime;
  }

  public BasicProcessTaskExecutor<T> getExecutor() {
    return executor;
  }

  public void setExecutor(BasicProcessTaskExecutor<T> executor) {
    this.executor = executor;
  }

  public ThreadPoolExecutor getTaskThreadPool() {
    return taskThreadPool;
  }

  public void setTaskThreadPool(ThreadPoolExecutor taskThreadPool) {
    this.taskThreadPool = taskThreadPool;
  }

  public DataGetType getDataGetType() {
    return dataGetType;
  }

  public void setDataGetType(DataGetType dataGetType) {
    this.dataGetType = dataGetType;
  }

}